package com.tycmc.storm.topology;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.storm.grouping.LoadAwareShuffleGrouping;
import org.apache.storm.topology.*;
import org.apache.storm.tuple.Fields;

import java.lang.reflect.Constructor;

/**
 * 拓扑工具类
 *
 * @author yangtao
 */
public class TopologyUtil {

    /**
     * 生成拓扑构建器
     *
     * @param componentArray - 拓扑结构字符串（json格式）
     * @return 拓扑构建器
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static TopologyBuilder generateTopologyBuilder(JSONArray componentArray) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        try {
            // 设置拓扑结构
            for (int i = 0; i < componentArray.size(); i++) {// 遍历拓扑的每个component(spout/bolt)
                JSONObject obj = componentArray.getJSONObject(i);
                //生成节点实例
                Class clz = Class.forName(obj.getString(TopologyConstant.COMPONENT_CLASSNAME));
                Object component;
                if (obj.get(TopologyConstant.COMPONENT_PARAM) != null) {//存在参数则调用构造器
                    Constructor constructor = clz.getConstructor(String.class); //构造函数参数列表的class类型
                    component = constructor.newInstance(obj.getString(TopologyConstant.COMPONENT_PARAM));
                } else {
                    component = clz.newInstance();
                }
                //将节点设置到拓扑构建器中
                if (obj.getString(TopologyConstant.COMPONENT_TYPE).equals(TopologyConstant.COMPONENT_TYPE_SPOUT)) {
                    SpoutDeclarer spoutDec = builder.setSpout(obj.getString("name"), (IRichSpout) component, obj.getIntValue(TopologyConstant.COMPONENT_PARALLEL));
                    if (obj.get(TopologyConstant.COMPONENT_TASKNUM) != null) {
                        spoutDec.setNumTasks(obj.getIntValue(TopologyConstant.COMPONENT_TASKNUM));
                    }
                    if (obj.get(TopologyConstant.COMPONENT_MAX_PENDING) != null) {
                        spoutDec.setMaxSpoutPending(obj.getIntValue(TopologyConstant.COMPONENT_MAX_PENDING));
                    }
                } else if (obj.getString(TopologyConstant.COMPONENT_TYPE).equals(TopologyConstant.COMPONENT_TYPE_BOLT)) {
                    BoltDeclarer bd = builder.setBolt(obj.getString(TopologyConstant.COMPONENT_NAME), (IRichBolt) component, obj.getIntValue(TopologyConstant.COMPONENT_PARALLEL));
                    if (obj.get(TopologyConstant.COMPONENT_TASKNUM) != null) {
                        bd.setNumTasks(obj.getIntValue(TopologyConstant.COMPONENT_TASKNUM));
                    }
                    if (obj.get(TopologyConstant.COMPONENT_MAX_PENDING) != null) {
                        bd.setMaxSpoutPending(obj.getIntValue(TopologyConstant.COMPONENT_MAX_PENDING));
                    }
                    JSONArray followArray = obj.getJSONArray(TopologyConstant.COMPONENT_FOLLOW);
                    for (int j = 0; j < followArray.size(); j++) {
                        String groupType = followArray.getJSONObject(j).getString(TopologyConstant.COMPONENT_FOLLOW_GROUPTYPE);
                        String target = followArray.getJSONObject(j).getString(TopologyConstant.COMPONENT_FOLLOW_TARGET);
                        String streamId = followArray.getJSONObject(j).getString(TopologyConstant.COMPONENT_FOLLOW_STREAMID);
                        if ("shuffle".equals(groupType)) {
                            if (streamId == null || streamId.isEmpty()) {
                                bd.shuffleGrouping(target);
                            } else {
                                bd.shuffleGrouping(target, streamId);
                            }
                        } else if ("custom".equals(groupType)) {
                            if (streamId == null || streamId.isEmpty()) {
                                bd.customGrouping(target, new LoadAwareShuffleGrouping());
                            } else {
                                bd.customGrouping(target, streamId, new LoadAwareShuffleGrouping());
                            }
                        } else if ("localOrShuffle".equals(groupType)) {
                            if (streamId == null || streamId.isEmpty()) {
                                bd.localOrShuffleGrouping(target);
                            } else {
                                bd.localOrShuffleGrouping(target, streamId);
                            }
                        } else if ("all".equals(groupType)) {
                            if (streamId == null || streamId.isEmpty()) {
                                bd.allGrouping(target);
                            } else {
                                bd.allGrouping(target, streamId);
                            }
                        } else if ("global".equals(groupType)) {
                            if (streamId == null || streamId.isEmpty()) {
                                bd.globalGrouping(target);
                            } else {
                                bd.globalGrouping(target, streamId);
                            }
                        } else if ("fields".equals(groupType)) {
                            String fieldName = followArray.getJSONObject(j).getString(TopologyConstant.COMPONENT_FOLLOW_FIELD);
                            if (streamId == null || streamId.isEmpty()) {
                                bd.fieldsGrouping(target, new Fields(fieldName));
                            } else {
                                bd.fieldsGrouping(target, streamId, new Fields(fieldName));
                            }
                        } else {
                            throw new Exception("不支持的分组类型：" + groupType);
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new Exception("生成拓扑构建器时发生异常,请检查拓扑结构字符串!", e);
        }
        return builder;
    }

}
